package spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.executor.task;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.tasks_manager.tasks_manager.node.Node;
import spring.cloud.tasks.tasks_manager.tasks_manager.service.AllotTaskItemService;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.TaskItemAllotTask;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.executor.Executor;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.task.TaskItem;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * 作业的executor上线，executor或者该作业级别平衡摘取，只摘取该作业的shard；添加的新的shard；
 */
@Slf4j
@Setter
@Getter
public class TaskExecutorOnlineTaskItemAllotTask extends TaskItemAllotTask {
    private String taskId;
    private String executorId;

    public TaskExecutorOnlineTaskItemAllotTask(AllotTaskItemService allotTaskItemService, String executorId, String taskId) {
        super(allotTaskItemService);
        this.taskId = taskId;
        this.executorId = executorId;
    }

    @Override
    protected void logStartInfo() {
        log.info("Execute the {}, jobName is {}, executorName is {}", this.getClass().getSimpleName(), taskId, executorId);
    }

    @Override
    protected boolean pick(List<String> taskIdList,//
                           List<String> enableTaskIdList,//
                           List<TaskItem> taskItemList,//
                           List<Executor> onlineExecutorList,//
                           List<Executor> onlineOwnTrafficExecutorList) throws Exception {
        // 很小的可能性：status的新增事件先于ip的新增事件
        // 那么，如果lastOnlineExecutorList不包含executorName，则添加一个新的Executor
        // 添加当前作业至jobNameList
        Executor targetExecutor = null;
        for (int i = 0; i < onlineExecutorList.size(); i++) {
            Executor executor = onlineExecutorList.get(i);
            if (executor.getExecutorId().equals(executorId)) {
                targetExecutor = executor;
                break;
            }
        }
        if (targetExecutor == null) {
            targetExecutor = new Executor();
            targetExecutor.setExecutorId(executorId);
            targetExecutor.setIp(getIp());//可能返回为空 代表Executor已经挂掉了 但是能够走到这代表该执行器刚刚上线 理论上不会为空
            targetExecutor.setLostTraffic(getExecutorLostTraffic(executorId));
            targetExecutor.setTaskIdList(new ArrayList<String>());
            targetExecutor.setTaskItemList(new ArrayList<TaskItem>());
            targetExecutor.setTotalLoadLevel(0);
            onlineExecutorList.add(targetExecutor);
            if (!targetExecutor.isLostTraffic()) {
                onlineOwnTrafficExecutorList.add(targetExecutor);
            }
        }
        if (!targetExecutor.getTaskIdList().contains(taskId)) {
            targetExecutor.getTaskIdList().add(taskId);
        }

        // 如果该Executor流量被摘取，则无需摘取，返回true
        if (targetExecutor.isLostTraffic()) {
            return true;
        }

        pickIntelligent(enableTaskIdList, taskItemList, onlineOwnTrafficExecutorList);

        return true;
    }

    /**
     * @return 可能返回为空
     * @throws Exception
     */
    private String getIp() throws Exception {
        String ip = null;
        String executorIpNodePath = Node.getExecutorIpNodePath(executorId);
        if (curatorFramework.checkExists().forPath(Node.getExecutorIpNodePath(executorId)) != null) {
            byte[] bytes = curatorFramework.getData().forPath(executorIpNodePath);
            if (bytes != null) {
                ip = new String(bytes, StandardCharsets.UTF_8);
            }
        }
        return ip;
    }

    public void pickIntelligent(List<String> enableTaskIdList, List<TaskItem> taskItemList, List<Executor> onlineOwnTrafficExecutorList) throws Exception {
        boolean preferExecutorIdListIsConfigured = preferExecutorIdListIsConfigured(taskId); // 是否配置了preferList
        List<String> preferExecutorIdListConfigured = getPreferExecutorIdListConfigured(taskId); // 配置态的preferList
        int loadLevel = getLoadLevel(taskId);
        int totalTaskItemCount = getTotalTaskItemCount(taskId);
        //
        boolean hasTaskItemRunningOfThisTaskId = hasTaskItemRunningOfThisTaskId(onlineOwnTrafficExecutorList);
        if (preferExecutorIdListIsConfigured) {
            pickIntelligentWithPreferExecutorIdListIsConfigured(//
                    enableTaskIdList, //
                    taskItemList, //
                    preferExecutorIdListConfigured, //
                    onlineOwnTrafficExecutorList,//
                    hasTaskItemRunningOfThisTaskId,//
                    totalTaskItemCount, //
                    loadLevel);//
        } else {
            pickIntelligentWithOutPreferExecutorIdListIsConfigured(//
                    enableTaskIdList, //
                    taskItemList, //
                    onlineOwnTrafficExecutorList,//
                    hasTaskItemRunningOfThisTaskId,//
                    totalTaskItemCount,//
                    loadLevel);
        }
    }


    private boolean hasTaskItemRunningOfThisTaskId(List<Executor> onlineOwnTrafficExecutorList) {
        for (int i = 0; i < onlineOwnTrafficExecutorList.size(); i++) {
            List<TaskItem> taskItemList = onlineOwnTrafficExecutorList.get(i).getTaskItemList();
            for (int j = 0; j < taskItemList.size(); j++) {
                if (taskItemList.get(j).getTaskId().equals(taskId)) {
                    return true;
                }
            }
        }
        return false;
    }

    //
    private void pickIntelligentWithPreferExecutorIdListIsConfigured(
            List<String> enableTaskIdList,//
            List<TaskItem> taskItemList,//
            List<String> preferExecutorIdListConfigured,//
            List<Executor> onlineOwnTrafficExecutorList,//
            //
            boolean hasTaskItemRunningOfThisTaskId,//
            int totalTaskItemCount,//
            int loadLevel) throws Exception {
        if (preferExecutorIdListConfigured.contains(executorId)) {
            // 如果有分片正在运行，摘取全部运行在非优先节点上的分片，还可以平衡摘取
            if (hasTaskItemRunningOfThisTaskId) {
                taskItemList.addAll(pickTaskItemListRunningInDisPreferExecutorIdList(preferExecutorIdListConfigured, onlineOwnTrafficExecutorList));
                if (ENABLE_TASK_BASED_SHARDING) {
                    pickBalanceWithTask(taskItemList, getOnlineExecutorListInPreferExecutorIdListConfigured(preferExecutorIdListConfigured, onlineOwnTrafficExecutorList, true));
                } else {
                    pickBalance(taskItemList, onlineOwnTrafficExecutorList);
                }
            } else {
                // 如果没有分片正在运行，则需要新建，无需平衡摘取
                if (enableTaskIdList.contains(taskId)) {
                    taskItemList.addAll(createTaskItemList(totalTaskItemCount, loadLevel));
                }
            }
        } else {
            boolean useDisPreferExecutorIdList = useDisPreferExecutorIdList(taskId);
            if (useDisPreferExecutorIdList) {
                // 如果有分片正在运行，并且都是运行在非优先节点上，可以平衡摘取分片
                // 如果有分片正在运行，并且有运行在优先节点上，则摘取全部运行在非优先节点上的分片，不能再平衡摘取
                if (hasTaskItemRunningOfThisTaskId) {
                    boolean taskItemAllRunningInDisPreferList = taskItemIsAllRunningInDisPreferList(preferExecutorIdListConfigured, onlineOwnTrafficExecutorList);
                    if (taskItemAllRunningInDisPreferList) {
                        if (ENABLE_TASK_BASED_SHARDING) {
                            pickBalanceWithTask(taskItemList, getOnlineExecutorListInPreferExecutorIdListConfigured(preferExecutorIdListConfigured, onlineOwnTrafficExecutorList, false));
                        } else {
                            pickBalance(taskItemList, onlineOwnTrafficExecutorList);
                        }
                    } else {
                        //为什么这里不进行平衡摘取 因为当前的节点不是preferExecutorId
                        taskItemList.addAll(pickTaskItemListRunningInDisPreferExecutorIdList(preferExecutorIdListConfigured, onlineOwnTrafficExecutorList));
                    }
                } else {
                    // 如果没有分片正在运行，则需要新建，无需平衡摘取
                    if (enableTaskIdList.contains(taskId)) {
                        taskItemList.addAll(createTaskItemList(totalTaskItemCount, loadLevel));
                    }
                }
            } else { // 不能再平衡摘取
                // 摘取全部运行在非优先节点上的分片
                taskItemList.addAll(pickTaskItemListRunningInDisPreferExecutorIdList(preferExecutorIdListConfigured, onlineOwnTrafficExecutorList));
            }
        }
    }

    private List<Executor> getOnlineExecutorListInPreferExecutorIdListConfigured(
            List<String> preferExecutorIdListConfigured,//
            List<Executor> onlineOwnTrafficExecutorList,//
            boolean contains) {//
        List<Executor> executorList = new ArrayList<>();
        for (Executor executor : onlineOwnTrafficExecutorList) {
            if (executor.getTaskIdList().contains(taskId) && contains == preferExecutorIdListConfigured.contains(executor.getExecutorId())) {
                executorList.add(executor);
            }
        }
        return executorList;
    }

    private List<TaskItem> pickTaskItemListRunningInDisPreferExecutorIdList(List<String> preferExecutorIdListConfigured, List<Executor> onlineOwnTrafficExecutorList) {
        List<TaskItem> taskItemList = new ArrayList<>();
        for (int i = 0; i < onlineOwnTrafficExecutorList.size(); i++) {
            Executor executor = onlineOwnTrafficExecutorList.get(i);
            if (!preferExecutorIdListConfigured.contains(executor.getExecutorId())) {
                Iterator<TaskItem> iterator = executor.getTaskItemList().iterator();
                while (iterator.hasNext()) {
                    TaskItem taskItem = iterator.next();
                    if (taskItem.getTaskId().equals(taskId)) {
                        iterator.remove();
                        executor.setTotalLoadLevel(executor.getTotalLoadLevel() - taskItem.getLoadLevel());
                        //
                        taskItemList.add(taskItem);
                    }
                }
            }
        }
        return taskItemList;
    }

    private boolean taskItemIsAllRunningInDisPreferList(List<String> preferExecutorIdListConfigured, List<Executor> onlineOwnTrafficExecutorList) {
        for (int i = 0; i < onlineOwnTrafficExecutorList.size(); i++) {
            Executor executor = onlineOwnTrafficExecutorList.get(i);
            if (preferExecutorIdListConfigured.contains(executorId)) {
                List<TaskItem> taskItemList = executor.getTaskItemList();
                for (int j = 0; j < taskItemList.size(); j++) {
                    if (taskItemList.get(j).getTaskId().equals(taskId)) {
                        return false;
                    }
                }
            }
        }
        return true;
    }

    private void pickIntelligentWithOutPreferExecutorIdListIsConfigured(//
                                                                        List<String> enableTaskIdList,//
                                                                        List<TaskItem> taskItemList,//
                                                                        List<Executor> onlineOwnTrafficExecutorList,//
                                                                        boolean hasTaskItemRunningOfThisTaskId,//
                                                                        int totalTaskItemCount,//
                                                                        int loadLevel) {//
        // 如果有分片正在运行，则平衡摘取
        if (hasTaskItemRunningOfThisTaskId) {
            if (ENABLE_TASK_BASED_SHARDING) {
                pickBalanceWithTask(taskItemList, getOnlineExecutorListOfTaskId(onlineOwnTrafficExecutorList));
            } else {
                pickBalance(taskItemList, onlineOwnTrafficExecutorList);
            }
        } else {
            // 如果没有分片正在运行，则需要新建，无需平衡摘取
            if (enableTaskIdList.contains(taskId)) {
                taskItemList.addAll(createTaskItemList(totalTaskItemCount, loadLevel));
            }
        }
    }

    // 过滤能运行该作业的优先节点，或者非优先节点
    private List<Executor> getOnlineExecutorListOfTaskId(List<Executor> onlineOwnTrafficExecutorList) {
        List<Executor> executorList = new ArrayList<>();
        for (Executor executor : onlineOwnTrafficExecutorList) {
            if (executor.getTaskIdList().contains(taskId)) {
                executorList.add(executor);
            }
        }
        return executorList;
    }

    private void pickBalanceWithTask(List<TaskItem> taskItemList, List<Executor> executorList) {
        // 计算出executor对该作业的平均负荷
        int totalLoadLevel = 0;
        for (TaskItem taskItem : taskItemList) {
            totalLoadLevel += taskItem.getLoadLevel();
        }
        for (Executor executor : executorList) {
            for (TaskItem taskItem : executor.getTaskItemList()) {
                if (taskId.equals(taskItem.getTaskId())) {
                    totalLoadLevel += taskItem.getLoadLevel();
                }
            }
        }
        int averageTotalLoadLevelAtPerExecutor = totalLoadLevel / executorList.size();
        // 摘取executor超出平均负荷的分片
        for (Executor executor : executorList) {
            int taskTotalLoadLevel = 0;
            Iterator<TaskItem> iterator = executor.getTaskItemList().iterator();
            while (iterator.hasNext()) {
                TaskItem taskItem = iterator.next();
                if (taskId.equals(taskItem.getTaskId())) {
                    if (averageTotalLoadLevelAtPerExecutor == 0 || taskTotalLoadLevel + taskItem.getLoadLevel() > averageTotalLoadLevelAtPerExecutor) {
                        iterator.remove();
                        executor.setTotalLoadLevel(executor.getTotalLoadLevel() - taskItem.getLoadLevel());
                        //
                        taskItemList.add(taskItem);
                    } else {
                        //不超过平均权重则保留。
                        taskTotalLoadLevel += taskItem.getLoadLevel();
                    }
                }
            }
        }
    }

    // 计算平均load，然后摘取最接近平均负载的shard。
    private void pickBalance(List<TaskItem> taskItemList, List<Executor> onlineOwnTrafficExecutorList) {
        int totalLoadLevel = getTotalLoadLevel(taskItemList, onlineOwnTrafficExecutorList);
        int averageTotalLoadLevelAtPerExecutor = totalLoadLevel / (onlineOwnTrafficExecutorList.size());
        for (Executor executor : onlineOwnTrafficExecutorList) {
            pickBalance(taskItemList, executor, averageTotalLoadLevelAtPerExecutor);
        }
    }

    private int getTotalLoadLevel(List<TaskItem> taskItemList, List<Executor> onlineOwnTrafficExecutorList) {
        int totalLoadLevel = 0;
        for (int i = 0; i < taskItemList.size(); i++) {
            totalLoadLevel += taskItemList.get(i).getLoadLevel();
        }
        for (int i = 0; i < onlineOwnTrafficExecutorList.size(); i++) {
            totalLoadLevel += onlineOwnTrafficExecutorList.get(i).getTotalLoadLevel();
        }
        return totalLoadLevel;
    }

    private void pickBalance(List<TaskItem> taskItemList, Executor executor, int averageTotalLoadLevelAtPerExecutor) {
        int needPickedLoadLevel = executor.getTotalLoadLevel() - averageTotalLoadLevelAtPerExecutor;
        while (needPickedLoadLevel > 0 && !executor.getTaskItemList().isEmpty()) {
            // 摘取现在totalLoad > 平均值的executor里面的shard
            TaskItem pickTaskItem = null;
            //下一次for循环重新计算taskItemList().size(),不会产生问题
            for (int j = 0; j < executor.getTaskItemList().size(); j++) {
                TaskItem taskItem = executor.getTaskItemList().get(j);
                if (!taskItem.getTaskId().equals(taskId)) { // 如果当前Shard不属于该作业，则不摘取，继续下一个
                    continue;
                }
                if (pickTaskItem == null) {
                    pickTaskItem = taskItem;
                    continue;
                }//第一次赋值
                //需要计算出比临界值大的里面取最小,比临界值小的里面取最大。
                //优先找比needPickedLoadLevel大的最接近的权重,因为摘取的权重够了就不用继续进行遍历摘取了。
                //大家会担心如果权重大于needPickedLoadLevel的大很多，其实一般权重不会设置很大的。
                if (pickTaskItem.getLoadLevel() >= needPickedLoadLevel) {
                    if (taskItem.getLoadLevel() >= needPickedLoadLevel && taskItem.getLoadLevel() < pickTaskItem.getLoadLevel()) {
                        pickTaskItem = taskItem;
                    } else {
                        //...保持什么操作也不做
                    }
                    continue;
                } else {
                    //这里pickTaskItem.getLoadLevel() < needPickedLoadLevel
                    //优先取比临界值大的
                    if (taskItem.getLoadLevel() >= needPickedLoadLevel) {
                        pickTaskItem = taskItem;
                        continue;
                    } else {
                        if (taskItem.getLoadLevel() > pickTaskItem.getLoadLevel()) {
                            pickTaskItem = taskItem;
                        }
                        continue;
                    }
                }
            }//end for
            //该值可能比临界值大也可能小
            if (pickTaskItem != null) {
                executor.getTaskItemList().remove(pickTaskItem);
                executor.setTotalLoadLevel(executor.getTotalLoadLevel() - pickTaskItem.getLoadLevel());
                //
                taskItemList.add(pickTaskItem);
            } else {
                // 没有符合摘取条件的，无需再选择摘取
                break;
            }
            needPickedLoadLevel = executor.getTotalLoadLevel() - averageTotalLoadLevelAtPerExecutor;
        }
    }

    private List<TaskItem> createTaskItemList(int totalTaskItemCount, int loadLevel) {
        List<TaskItem> taskItemList = new ArrayList<>();
        for (int taskItemIndex = 0; taskItemIndex < totalTaskItemCount; taskItemIndex++) {
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskId(taskId);
            taskItem.setTaskItemIndex(taskItemIndex);
            taskItem.setLoadLevel(loadLevel);
            taskItemList.add(taskItem);
        }
        return taskItemList;
    }


    //////////////////////////////////////////////////////////////////////////////////////////////////


}
